package com.bigdata.spark.mallapp_realtime

import java.lang

import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import com.bigdata.spark.util.{MyKafkaUtil, OffsetManagerUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable.ListBuffer

object mall_log_PagesViewTop_count {

  //统计页面浏览量。每30秒统计各种页面浏览量。

  def main(args: Array[String]): Unit = {
    val actions_count: SparkConf = new SparkConf().setMaster("local[*]").setAppName("actions_count")
    val ssc: StreamingContext = new StreamingContext(actions_count,Seconds(5))


    var topic:String = "gmall_event_0523"
    var groupId = "PV_TOP"

    val offsetMap: Map[TopicPartition, Long] = OffsetManagerUtil.getOffset(topic, groupId)

    var recordDStream: InputDStream[ConsumerRecord[String, String]] = null

    if(offsetMap.size>0&& offsetMap!=null){
      recordDStream = MyKafkaUtil.getKafkaStream(topic,ssc,offsetMap,groupId)
    }else{
      recordDStream = MyKafkaUtil.getKafkaStream(topic,ssc,groupId)
    }


    var offsetRanges: Array[OffsetRange] = Array.empty[OffsetRange]

    val KafkaDS: DStream[ConsumerRecord[String, String]] = recordDStream.transform {
      rdd => {
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }

    }

    def getpage(Kafkadata:ConsumerRecord[String, String]) = {
      val str: String = Kafkadata.value()
      val jsonlog: JSONObject = JSON.parseObject(str)
      val page: JSONObject = jsonlog.getJSONObject("page")
      val common: JSONObject = jsonlog.getJSONObject("common")
      (page,common)
    }


    val LPRDD: DStream[((String, String),Int)] = KafkaDS.map(getpage).map(
      pageandcommon => {
        ((pageandcommon._1.getString("page_id"), pageandcommon._2.getString("ar")),1)
      }
    )

    val windowDS: DStream[((String, String),Int)] = LPRDD.window(Seconds(30), Seconds(30))


    val resRDD: DStream[((String, String), Int)] = windowDS.reduceByKey(_ + _)



    resRDD.foreachRDD{
      rdd=>{

        rdd.foreach(
          record=>{
            println(record)//完成。这里用动作id+地区id作为主键，把最近访问次数统计放入redis里就好了
          }
        )

        println("第一层")
        println(rdd,rdd.getClass)
        //提交偏移量到Redis中，以分区为单位。
        //println("提交偏移量到Redis中，以分区为单位。")
        OffsetManagerUtil.saveOffset(topic,groupId,offsetRanges)

      }

    }


    ssc.start()

    ssc.awaitTermination()




  }
}
